29 实践课-多轮对话优化与场景落地

多轮对话优化与场景落地

关联：索引

术语小抄（初学者版）

课程思政融入点（口径统一）：

1）历史丢失（系统没“带上关键历史”）

2）指代错误（系统“带了历史但绑错了”）

  1. 先看 session_id 与 task_state 是否正确更新（状态是否可复现）。
  2. 再看“上下文关联规则”：本轮输入到底喂了哪些历史与证据（是否漏了关键字段/喂了无关内容）。
  3. 最后看“解析与生成”：是解析槽位错（parse_trace_id）还是回复组织/执行错（trace_id）。

目标：把“感觉不对”变成“可记录、可统计、可回归”的问题单。

1）定义对话日志（JSONL，一行一条对话轮次）

{"session_id":"demo","turn_id":"t1","ts_ms":1710000000000,"user_text":"把苹果放到A箱","task_state_before":{},"assistant_text":"已理解：把苹果放到A箱（trace_id=demo1）","parse_trace_id":"p1","trace_id":"r1","slots":{"action":"put","item_name":"苹果","bin":"A"},"issue":null}
{"session_id":"demo","turn_id":"t2","ts_ms":1710000001000,"user_text":"上一个指令再来一次","task_state_before":{"last_item":"苹果","last_action":"put","last_bin":"A","last_raw_user":"把苹果放到A箱"},"assistant_text":"已理解：把苹果放到A箱（trace_id=demo2）","parse_trace_id":"p2","trace_id":"r2","slots":{"repeat_of":"把苹果放到A箱","action":"put","item_name":"苹果","bin":"A"},"issue":null}
{"session_id":"demo","turn_id":"t3","ts_ms":1710000002000,"user_text":"它改放到B箱","task_state_before":{"last_item":"苹果","last_action":"put","last_bin":"A","last_raw_user":"上一个指令再来一次"},"assistant_text":"需要澄清：你说的“它”指的是哪个物品？（trace_id=demo3）","parse_trace_id":"p3","trace_id":"r3","slots":{"need_clarification":true,"bin":"B","action":"put"},"issue":{"type":"COREF_ERROR","note":"task_state_before 已有 last_item，但指代消解仍触发澄清，说明“状态未被正确使用/规则未命中”"}}
{"session_id":"demo","turn_id":"t4","ts_ms":1710000003000,"user_text":"这个不要了，换成香蕉","task_state_before":{"last_item":"苹果","last_action":"put","last_bin":"A","last_raw_user":"把苹果放到A箱"},"assistant_text":"已理解：替换物品为香蕉，将沿用上一条动作/箱位（trace_id=demo4）","parse_trace_id":"p4","trace_id":"r4","slots":{"ref_item":"苹果","action":"replace","item_name":"香蕉"},"issue":null}

解释与自检要点：

2）实现问题归类脚本 analyze_dialogue_issues.py

# analyze_dialogue_issues.py
# 目标：把多轮对话“哪里不好用”变成可统计的结构化问题类别，便于：
# - 课堂快速定位：历史丢失 vs 指代错误 vs 合理追问
# - 工程回归：用同一套分类规则对比“修复前/修复后”的问题分布
# - 写作业：输出 issue_counts 与典型 examples（可截图/可复盘）
import json
from typing import Any, Dict, List, Tuple

def _as_dict(x: Any) -> Dict[str, Any]:
    # 容错：确保后续对 x.get(...) 的调用不会崩溃
    # - 若 x 不是 dict，则当作空 dict（表示“没有状态/没有槽位”）
    return x if isinstance(x, dict) else {}

def classify_issue(turn: Dict[str, Any]) -> Tuple[str, str]:
    # 输入：一轮对话的结构化记录（来自 dialogue_turns.jsonl 的一行）
    # 输出：(issue_type, note)
    # - issue_type：用于统计聚合（例如 HISTORY_LOSS / COREF_ERROR）
    # - note：用于展示说明（便于学生理解为什么被归类）
    user_text = str(turn.get("user_text", "")).strip()
    slots = _as_dict(turn.get("slots"))
    state = _as_dict(turn.get("task_state_before"))

    if not str(turn.get("session_id", "")).strip():
        # session_id 是多轮对话的“绑定键”
        # - 缺失会导致不同用户/不同任务串台，属于高优先级问题
        return "SESSION_MISSING", "session_id 为空，可能导致多会话串台"

    if "它" in user_text or "这个" in user_text or "上一个" in user_text:
        # 只要出现典型指代词，就进入“指代/回放”类诊断分支
        if slots.get("need_clarification") is True:
            # need_clarification=True：系统选择“追问澄清”
            # 追问不一定是错，但如果 task_state 里已经有足够信息仍追问，
            # 说明：状态未被正确使用 / 未传入解析模块 / 规则未命中
            if "上一个" in user_text and not str(state.get("last_raw_user", "")).strip():
                # “上一个指令”需要 last_raw_user（上一轮原始用户指令）才能回放
                return "HISTORY_LOSS", "缺 last_raw_user，无法回放上一个指令"
            if ("它" in user_text or "这个" in user_text) and not str(state.get("last_item", "")).strip():
                # “它/这个”需要 last_item（上一轮物品）才能绑定指代
                return "HISTORY_LOSS", "缺 last_item，无法绑定指代"
            if ("它" in user_text or "这个" in user_text) and str(state.get("last_item", "")).strip():
                # state 已有 last_item 却仍追问：更像“指代规则没用上”
                return "COREF_ERROR", "task_state 已有 last_item，但仍触发澄清：可能是指代规则未命中/状态未传入解析模块"
            if "上一个" in user_text and str(state.get("last_raw_user", "")).strip():
                # state 已有 last_raw_user 却仍追问：更像“回放规则没用上”
                return "COREF_ERROR", "task_state 已有 last_raw_user，但仍触发澄清：可能是回放规则未命中/状态未传入解析模块"
            # 其它追问：可能是合理行为（例如用户信息确实不足）
            return "NEED_CLARIFY", "触发澄清追问（可能是合理行为）"

        if ("它" in user_text or "这个" in user_text) and slots.get("ref_item"):
            # ref_item：被指代的“旧实体”（例如“这个不要了”里的“这个”）
            # item_name：当前轮最终要执行的实体（例如“换成香蕉”里的“香蕉”）
            # action：本轮动作（put/query_rule/replace/cancel 等）
            ref_item = str(slots.get("ref_item", "")).strip()
            item_name = str(slots.get("item_name", "")).strip()
            action = str(slots.get("action", "")).strip()

            if ("换成" in user_text or action == "replace") and ref_item and item_name and item_name == ref_item:
                # replace 场景里，新实体与旧实体不应相同
                # 若相同，常见原因是“过度替换/新实体丢失”，属于指代错误
                return "COREF_ERROR", "replace 场景新旧实体被覆盖为同一值，疑似“新实体丢失/过度替换”"

            if action in ("put", "query_rule") and ref_item and item_name and item_name == ref_item:
                # pronoun 场景里，item_name 继承 ref_item 是合理承接
                # 例如：“把它放到B箱”应继承上一轮的物品名
                return "OK", "pronoun 场景：item_name 继承 ref_item，属于正常指代承接"

    if slots.get("action") in ("put", "replace") and not slots.get("item_name"):
        # 控制类动作（put/replace）缺 item_name 风险较高：容易误操作
        # 工业场景建议：追问澄清或拒绝执行（门禁）
        return "SLOT_MISSING", "控制类动作缺 item_name，建议追问或拒绝执行"

    # 默认：未发现明显问题（或该问题不在本脚本的最小覆盖范围）
    return "OK", "未发现明显问题"

def main() -> None:
    # 约定：dialogue_turns.jsonl 与脚本同目录
    # - 一行一条 JSON（不要跨行），便于追加/定位/筛选
    path = "./dialogue_turns.jsonl"
    counts: Dict[str, int] = {}
    samples: Dict[str, List[Dict[str, Any]]] = {}

    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            # 逐行解析：每行必须是一个完整 JSON 对象
            turn = json.loads(line)
            t, note = classify_issue(turn)
            counts[t] = counts.get(t, 0) + 1
            if t != "OK":
                # 只收集非 OK 的样例，便于课堂展示与写“问题修复记录”
                samples.setdefault(t, []).append(
                    {
                        "turn_id": turn.get("turn_id"),
                        "user_text": turn.get("user_text"),
                        "task_state_before": turn.get("task_state_before"),
                        "slots": turn.get("slots"),
                        "note": note,
                    }
                )

    # 统计输出：先按数量降序，便于定位“最常见的问题”
    print("issue_counts:", dict(sorted(counts.items(), key=lambda x: (-x[1], x[0]))))
    for t in sorted(samples.keys()):
        print("----", t, "examples:")
        for s in samples[t][:3]:
            print(s)

if __name__ == "__main__":
    # 运行方式：py -3 .\analyze_dialogue_issues.py
    main()

逐段解释与自检要点：

3）运行自测（PowerShell）

py -3 .\analyze_dialogue_issues.py

解释与自检要点：

目标：用“扩充标注数据 + 调整规则”实现可量化提升，而不是靠感觉“越改越乱”。

1）扩充多轮标注数据（建议模板）

2）实现“上下文关联规则”配置 context_policy.py

# context_policy.py
# 目标：把“每轮到底带哪些上下文”变成可配置策略，而不是写死在业务代码里。
# 好处：
# - 可 A/B 测试：改 max_turns 或字段列表即可比较效果
# - 可回归：配合标注数据验证“策略调整是否减少失败样例”
# - 可审计：上下文输入可落盘复现，排障更快
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

@dataclass(frozen=True)
class ContextPolicy:
    # max_turns：短期对话窗口上限（只截取最近 N 轮）
    # - 控制 token 成本与延迟
    # - 避免历史无限增长导致“上下文污染”
    max_turns: int = 10
    # include_task_state_keys：从长期状态里挑“必须带入”的关键字段
    # - 只带关键字段，不带整份长期记忆，避免污染与泄露
    include_task_state_keys: List[str] = field(
        default_factory=lambda: ["last_item", "last_bin", "last_action", "last_raw_user"]
    )
    # include_evidence_keys：工具证据/摘要字段（建议只放可验证的结果摘要）
    # - 目的：减少模型“凭空编造”，并让输出可追溯
    include_evidence_keys: List[str] = field(
        default_factory=lambda: ["last_tool", "last_tool_ok", "last_tool_summary"]
    )

def build_context_payload(
    *,
    user_text: str,
    recent_turns: List[Dict[str, Any]],
    task_state: Dict[str, Any],
    policy: Optional[ContextPolicy] = None,
) -> Dict[str, Any]:
    # policy=None 时使用默认策略，保证“最小可用且可运行”
    p = policy or ContextPolicy()

    # recent_turns：只保留最近 max_turns 轮，避免无限增长
    turns = recent_turns[-int(p.max_turns) :] if recent_turns else []

    # 从 task_state 中按白名单抽取字段，形成“上下文状态区”
    state_payload: Dict[str, Any] = {}
    for k in p.include_task_state_keys:
        v = task_state.get(k)
        if v not in (None, "", [], {}):
            state_payload[k] = v

    # 从 task_state 中按白名单抽取工具证据字段，形成“证据区”
    evidence_payload: Dict[str, Any] = {}
    for k in p.include_evidence_keys:
        v = task_state.get(k)
        if v not in (None, "", [], {}):
            evidence_payload[k] = v

    # 统一返回结构：便于落盘、复现、调试、对齐提示词/模型输入
    return {
        "user_text": user_text,
        "recent_turns": turns,
        "task_state": state_payload,
        "evidence": evidence_payload,
    }

逐段解释与自检要点：

3）把策略接入对话入口（对齐 28 的 DialogueManager.handle 思路）

集成要点（示例为伪代码，工程里替换成你们的 Agent 调用即可）：

handle(user_text):
  1) load task_state + recent_turns
  2) context_payload = build_context_payload(...)
  3) 用 context_payload 做解析与回复生成（规则或 LLM）
  4) 写入 short_term（user/assistant）
  5) 回写 long_term：只写关键字段（last_* + last_tool_summary）

自检要点：

2）在 ContextPolicy.include_task_state_keys 里增加/删除 1 个字段，观察对某一类样本的影响。
提示：把 last_raw_user 去掉通常会让“上一个指令”类样本变差。

把“问题归类 + 上下文关联规则”接入你们的分拣智能体：

课程思政融入点（口径统一）：

1）简洁性：一句话说结论 + 一句话说下一步

2）指令明确性：动作/对象/位置/约束必须齐

3）可追踪：每轮必须有追踪字段

4）可回退：错误风险场景必须可撤销/可更正

目标：把回复格式固定住，减少“模型输出漂移”，并让一线操作更快更稳。

1）实现回复格式化函数 reply_format.py

# reply_format.py
# 目标：把工业场景的对话输出固定成统一结构，减少“模型输出漂移”，便于系统对接与回归验证。
# 设计原则：
# - summary：给人看的最短结论
# - next_step：下一步可执行指令/追问（减少来回沟通）
# - trace_id/parse_trace_id：证据字段（可追踪、可排障、可复验）
# - slots：给工程看的结构化信息（可直接用于工具调用/路由）
from __future__ import annotations

import json
from typing import Any, Dict, Optional

def format_industrial_reply(
    *,
    ok: bool,
    trace_id: str,
    parse_trace_id: str,
    summary: str,
    next_step: Optional[str] = None,
    slots: Optional[Dict[str, Any]] = None,
    warnings: Optional[list] = None,
) -> str:
    # payload 是对外输出的“统一契约”
    # - 使用 dict 组织，再 json.dumps，避免手拼字符串出错
    payload: Dict[str, Any] = {
        # ok：本轮处理是否成功（注意：并不等价于业务动作已执行）
        "ok": bool(ok),
        # trace_id：对话链路追踪（定位“执行/生成/路由”问题）
        "trace_id": str(trace_id),
        # parse_trace_id：解析链路追踪（定位“理解/槽位抽取”问题）
        "parse_trace_id": str(parse_trace_id),
        # summary：给用户看的最短结论（工业现场强调简洁）
        "summary": str(summary).strip(),
        # next_step：下一步指令或澄清问题（为空则输出空字符串）
        "next_step": (str(next_step).strip() if next_step else ""),
        # slots：结构化槽位，供系统端复用（非 dict 则回退为空 dict）
        "slots": (slots if isinstance(slots, dict) else {}),
        # warnings：非致命问题集合（保留证据但不阻断主流程）
        "warnings": (warnings if isinstance(warnings, list) else []),
    }
    # ensure_ascii=False：保证中文可读；默认会输出 UTF-8 字符
    return json.dumps(payload, ensure_ascii=False)

逐段解释与自检要点：

2）示例输出（对一线人员友好）

{"ok":true,"trace_id":"r8f2a1c3","parse_trace_id":"p0a1b2c3","summary":"已理解：把苹果放到B箱","next_step":"如需更正，请说：改放到X箱 / 取消上一步","slots":{"action":"put","item_name":"苹果","bin":"B"},"warnings":[]}

自检要点：

目标：让 AI 的输出变成可交付的工程改动，而不是“看起来很对”。

1）给 AI 的指令模板：生成诊断清单与改进建议

你是资深对话系统工程师。请基于以下信息输出一个“可落地”的多轮对话优化方案：
1) 当前系统的对话问题统计（issue_counts、fail_modes、by_type）
2) 代表性失败样例（bad_cases，含 task_state_before、input、expected_slots、pred_slots）

输出要求：
- 先给“问题诊断清单”（按严重程度排序：误操作风险 > 频率 > 修复成本）
- 再给“最小改动方案”（只能修改：上下文关联规则 / 指代消解规则 / 澄清策略）
- 每条改动都要说明：会影响哪些样本类型、如何用标注数据回归验证
- 不要引入新第三方依赖

说明与自检要点：

2）给 AI 的指令模板：生成上下文理解增强代码（需可审计）

你是资深 Python 工程师。请在不引入第三方依赖的前提下，给出对以下函数的最小补丁：
- build_context_payload(...)
- resolve_sorting_references(...)

已知问题：
1) pronoun 类样本中，用户说“它改放到B箱”会触发不必要追问
2) replace 类样本中，“这个不要了，换成香蕉”偶发把新实体覆盖掉

输入：失败样例（含 task_state_before、input、expected_slots、pred_slots）
输出：
- 修改后的完整函数代码（可直接复制）
- 每处修改的理由
- 用 evaluate_coref.py 回归验证的预期变化（指标或失败样例减少）

说明与自检要点：

总分 10 分，建议 ≥8 分视为“可落地候选”，<8 分需给出改进项与下一轮验证计划。

2）苹果分拣对话样例（参考，学生可替换为本地产业口径）

样例 A（指代承接）：

样例 B（纠错回退）：

样例 C（澄清追问）：

自检要点：

大模型任务（可直接复制的指令模板）

作业：布置

1）提交优化后的多轮对话代码、问题修复记录。
要求：至少包含 1 个“修复前失败样例”与“修复后通过样例”，并说明根因与改动点。

2）提交场景适配性验证报告（含 3 组产业场景对话示例、适配性评分），300 字左右。
要求：每组示例至少 3 轮对话；评分需包含 5 个维度（每项 0–2）与总分，并写 1 条改进建议。

3）提交标注数据扩充与优化的过程记录，说明优化前后的效果对比。
要求：提供新增样本数量、by_type 指标截图（或文本输出），并描述至少 1 个失败模式如何被修复（或如何缩小影响范围）。

Markdown 与代码自检清单（提交前必做）